package com.wfy.orchestra.dispatch

import android.util.Log
import com.wfy.orchestra.monitor.DefaultPerfMonitor
import com.wfy.orchestra.monitor.PerformanceMonitor
import com.wfy.orchestra.thread.ITaskExecutor
import com.wfy.orchestra.thread.MainThreadExecutor
import com.wfy.orchestra.thread.PoolTaskExecutor
import java.util.LinkedList
import java.util.PriorityQueue
import java.util.Queue
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit
import java.util.concurrent.locks.ReentrantLock
import kotlin.concurrent.Volatile
import kotlin.concurrent.withLock

/**
 * @author : wangfengyuan
 * e-mail  : 446950883@qq.com
 * time    : 2025/3/19
 * desc    : 轮询优先级队列
 * version : 1.0
 */
open class DefaultTaskGraph(
    private val defaultExecutor: ITaskExecutor = PoolTaskExecutor(),
    private val mainExecutor: ITaskExecutor = MainThreadExecutor()
) : ITaskGraph {
    private val TAG = "TaskGraph"

    //单线程池，用于持续处理任务队列
    private val schedulerExecutor = Executors.newSingleThreadExecutor()

    //线程安全的哈希表，存储所有任务节点（Key: 任务ID，Value: TaskNode）
    private val taskNodes: MutableMap<String, TaskNode> = ConcurrentHashMap()

    // 任务状态追踪
    private val runningTasks = ConcurrentHashMap<TaskNode, Boolean>()

    //优先级比较器（数值大的优先级高）
    private val taskComparator = Comparator<TaskNode> { node1, node2 ->
        node2.getPriority().compareTo(node1.getPriority())
    }

    //优先级队列（存储待执行任务）和同步锁
    private val priorityQueue = PriorityQueue(11, taskComparator)

    // 使用Condition优化唤醒机制
    private val lock = ReentrantLock()// 可重入锁
    private val condition = lock.newCondition()// 条件变量

    @Volatile
    private var isCancelled = false //任务取消标志（volatile 保证多线程可见性）
    private var perfMonitor: PerformanceMonitor = DefaultPerfMonitor() //性能监控器
    private var errorListener: OnErrorListener? = null //错误监听器

    /**
     * 添加任务到图中
     */
    override fun addTask(task: ITask) {
        val v = taskNodes[task.id]
        if (v == null) {
            taskNodes[task.id] = TaskNode(task)
        }
    }

    override fun addTask(vararg tasks: ITask) {
        tasks.forEach { addTask(it) }
    }

    /**
     * 添加任务依赖关系
     */
    override fun addDependency(taskId: String, dependencyId: String) {
        val taskNode = taskNodes[taskId] ?: throw IllegalArgumentException("ITask with ID $taskId does not exist")
        val depNode = taskNodes[dependencyId] ?: throw IllegalArgumentException("Dependency with ID $dependencyId does not exist")
        depNode.addSuccessor(taskId)// 将taskId任务添加为dependencyId任务的后继
        taskNode.incrementInDegree()// 增加当前任务的入度
    }

    override fun addDependency(taskId: String, dependencyIds: List<String>) {
        dependencyIds.forEach { addDependency(taskId, it) }
    }

    override fun configureDependencies(block: TaskDslBuilder.() -> Unit) {
        TaskDslBuilder(this).apply(block)
    }

    override fun start() {
        Log.v(TAG, "taskNodes $taskNodes")//循环依赖时StackOverflowError,successors改为string可以解决
        check(!hasCycle()) { "Cycle detected in task graph" } // 检查循环依赖
        // 将入度为0的初始任务加入优先级队列
        lock.withLock {
            taskNodes.values.filter { it.getInDegree() == 0 }.forEach {
                priorityQueue.add(it)
            }
            // 唤醒调度线程，理论上启动时还无等待的调度线程，此处为兜底的冗余设计
            condition.signalAll()
        }
        Log.v(TAG, "priorityQueue $priorityQueue")
        // 启动持续调度
        schedulerExecutor.execute(::processPriorityQueue)
    }

    private fun processPriorityQueue() {
        while (!isCancelled) {
            lock.withLock {
                val node = priorityQueue.poll()
                if (node != null) {
                    submitTask(node)
                } else {
                    // 队列空时短暂等待，避免 CPU 忙等
                    condition.await(100, TimeUnit.MILLISECONDS)
                }
            }
        }
    }

    private fun submitTask(node: TaskNode) {
        if (isCancelled) {// 检查全局取消状态
            mainExecutor.execute { node.task.onCancelled() }
            return
        }
        runningTasks[node] = true
        val executor:ITaskExecutor = if (node.task.getExecutor() != null) node.task.getExecutor()!! else defaultExecutor
        perfMonitor.onTaskSubmit(node.task)
        executor.execute {
            //性能监控
            perfMonitor.onTaskStarted(node.task)
            try {
                // 双重检查
                if (!isCancelled) {
                    node.task.execute()
                }
            } catch (e: Exception) {
                //错误处理机制
                Log.e(TAG,"${node.task} execute error ${e.message}")
                if (errorListener != null) {
                    mainExecutor.execute {
                        errorListener!!.onError(node.task, e)
//                        cancelDependentTasks(node) //取消依赖此任务的后继
                    }
                }
                isCancelled = true
            } finally {
                perfMonitor.onTaskFinished(node.task)
                runningTasks.remove(node)
                // 确保生命周期销毁后不回调主线程
                if (!isCancelled) {
                    mainExecutor.execute { onTaskCompleted(node) }
                }
            }
        }
    }

    // 公开的取消方法
    override fun cancelAll() {
        isCancelled = true
        schedulerExecutor.shutdownNow() // 强制终止调度线程
        defaultExecutor.shutdown()
        lock.withLock {
            priorityQueue.clear()// 清空队列
            taskNodes.values.forEach { it.task.onCancelled() }// 取消所有任务
            taskNodes.clear()// 清空任务节点
            condition.signalAll()// 唤醒所有等待线程
        }
    }

    private fun onTaskCompleted(node: TaskNode) {
        lock.withLock {
            node.getSuccessors().forEach { successor ->
                // 减少后继任务的入度，若入度为0则加入队列
                if (taskNodes[successor]!!.decrementInDegree() == 0 && !isCancelled) {
                    priorityQueue.add(taskNodes[successor])
//                    condition.signal() // 精确唤醒
                    // 在任务入队时根据队列长度选择唤醒方式
                    if (priorityQueue.size == 1) condition.signal()
                    else condition.signalAll()
                }
            }
        }
    }

    /**
     * 循环依赖检测
     */
    private fun hasCycle(): Boolean {
        // 使用普通队列即可（拓扑排序不依赖优先级）
        val queue: Queue<TaskNode> = LinkedList()
        // 入度表
        val inDegreeMap: MutableMap<TaskNode, Int> = HashMap()

        // 初始化入度表
        for (node in taskNodes.values) {
            inDegreeMap[node] = node.getInDegree()
            if (inDegreeMap[node] == 0) {
                queue.add(node)
            }
        }

        var processedCount = 0
        while (!queue.isEmpty()) {
            val node = queue.poll()
            processedCount++
            // 减少所有后继的入度
            for (successor in node.getSuccessors()) {
                val successorTask = taskNodes[successor]
                val newInDegree = inDegreeMap[successorTask]!! - 1
                inDegreeMap[successorTask!!] = newInDegree
                if (newInDegree == 0) {
                    queue.add(successorTask)
                }
            }
        }
        // 如果处理的任务数 ≠ 总任务数 → 存在环
        return processedCount != taskNodes.size
    }

    //依赖可视化,生成DAG的Graphviz描述
    override fun generateDotGraph(): String {
        val sb = StringBuilder("digraph G {\n")
        taskNodes.values.forEach { node ->
            node.getSuccessors().forEach {
                sb.append("  \"${node.task.id}\" -> \"${it}\";\n")
            }
        }
        return sb.append("}").toString()
    }

    override fun updatePriority(taskId: String, newPriority: Int) {
        taskNodes[taskId]?.let { node ->
            lock.withLock {
                priorityQueue.remove(node)
                node.task.priority = newPriority // 需修改Task为可变优先级
                priorityQueue.add(node)
                condition.signal()
            }
        }
    }

    override fun setOnErrorListener(listener: OnErrorListener) {
        this.errorListener = listener
    }

    override fun setPerformanceMonitor(monitor: PerformanceMonitor) {
        this.perfMonitor = monitor
    }
}